#! /usr/bin/python2.5
# This file is part of SyncStar, free software released under GPL v2.

"""Unit tests for webapp/sms.py running on the AppEngine development server.

To run this test, please start `dev_appserver.py webapp' first. Don't run
multiple instances of this test in parallel, and don't make requests with the
test@unittest.org test user account while running the test.

TODO: Speed this up: Ran 8 tests in 27.691s
"""

import logging
import re
import os
import socket
import sys
import unittest
import zlib

import simplejson as json

class Response(object):
  __slots__ = ['status', 'status_message', 'content_type', 'charset', 'body',
               'json', 'redirect_target', 'is_compressed', 'compressed_size']

  def __init__(self, dict_obj=None):
    for name in self.__slots__:
      setattr(self, name, None)
    if dict_obj:
      for name in sorted(dict_obj):
        setattr(self, name, dict_obj[name])

  def to_dict(self):
    return dict((name, getattr(self, name)) for name in self.__slots__)

  def __repr__(self):
    return 'Response(%r)' % self.to_dict()


class NoJsonClass(object):
  """A token for Request.json."""

  def __repr__(self):
    return 'NoJson'

NoJson = NoJsonClass()


def simplify_url(url, address):
  match = re.match(r'http://([^:/?#\s]+)(?::(\d+))?/?', url)
  if match:
    host = match.group(1)
    if match.group(2) is None:
      port = 80
    else:
      port = match.group(2)
    if (address[0] == host or
        (address[0] in ('', '0', '0.0.0.0', '127.0.0.1', 'localhost',
                        'localhost.localdomain') and
         host in ('127.0.0.1', 'localhost', 'localhost.localdomain'))):
      return '/' + url[match.end():]
  return url


def recursive_unicode_to_utf8(obj):
  if isinstance(obj, unicode):
    return obj.encode('UTF-8')
  elif isinstance(obj, list):
    if type(obj) == list:
      return map(recursive_unicode_to_utf8, obj)
    else:
      return type(obj)(recursive_unicode_to_utf8(x) for x in obj)
  elif isinstance(obj, tuple):
    return type(obj)(recursive_unicode_to_utf8(x) for x in obj)
  elif isinstance(obj, dict):
    return type(obj)((recursive_unicode_to_utf8(name),
                      recursive_unicode_to_utf8(obj[name])) for name in obj)
  else:
    return obj


MESSAGE_FIELD_NAMES = ('address', 'body', 'date', 'type', 'service_center',
                       'error_code', 'locked', 'read', 'seen', 'protocol', 
                       'phone', 'deleted', 'status', 'conversation')


def assert_full_message(message):
  assert isinstance(message, dict)
  for name in MESSAGE_FIELD_NAMES:
    assert name in message
  assert len(message) == len(MESSAGE_FIELD_NAMES)


def simplify_message(message):
  """Returns a new, simplified message dict."""
  assert isinstance(message, dict)
  simplified_message = {}
  for name in ('address', 'body', 'date', 'type', 'service_center',
               'error_code', 'locked', 'read', 'seen', 'protocol', 
               'deleted'):
    if message[name]:  # 0, False or '' is the default.
      simplified_message[name] = message[name]
  if message['status'] != -1:
    simplified_message['status'] = message['status']
  if message['phone'] != 'test':
    simplified_message['phone'] = message['phone']
  if (message['conversation'] and
      message['conversation'] != message['address']):
    simplified_message['conversation'] = message['conversation']
  return simplified_message


def ph(message):
  assert isinstance(message, dict)
  message = dict(message)
  #if message.get('phone') == 'test':
  #  message.pop('phone')
  message.setdefault('phone', 'test')
  return message


class Client(object):
  __slots__ = ('address')

  def __init__(self):
    hostport = os.getenv('SMS_DEV_SERVER_HOSTPORT', '127.0.0.1:8080')
    match = re.match(r'(?:http://)?([^:/?#\s]+)(?::(\d+))?(?:/|\Z)', hostport)
    assert match, 'Bad server hostport: %r' % hostport
    host = match.group(1)
    if match.group(2) is None:
      port = 80
    else:
      port = int(match.group(2))
    self.address = (host, port)

  def check_health(self):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    try:
      try:
        sock.connect(self.address)
        sock.sendall('GET / HTTP/1.0\r\n\r\n')
        buf = []
        while True:
          data = sock.recv(8192)
          if not data:
            break
          buf.append(data)
        buf = ''.join(buf)
      except socket.error, e:
        assert 0, 'socket error to %r: %r' % (self.address, e)
    finally:
      sock.close()
    match = re.match(r'HTTP/1[.][01] +200 +OK\r?\n', buf)
    assert match, 'Expected successful HTTP response: %r' + buf[:128]

  def call(self, suburl, post_data=None, post_json=None, content_type=None,
           do_deflate=False):
    """Send HTTP request and return Response object."""
    # TODO: Add a faster version of this, first using HTTP keep-alive, then
    # avoiding HTTP and calling WSGI methods within the dev_appserver process
    # directly.
    assert (post_data is not None) + (post_json is not None) <= 1
    assert isinstance(suburl, str)
    assert suburl.startswith('/')
    content_type = None
    if post_json is not None:
      # TODO: Test compressed JSON.
      post_data = json.dumps(
          post_json, ensure_ascii=False, encoding='UTF-8',
          separators=(',', ':')).encode('UTF-8')
      del post_json
      content_type = 'application/json; charset=UTF-8'
      method = 'POST'
    elif post_data is not None:
      if content_type is None:
        content_type = 'application/x-www-form-urlencoded; charset=UTF-8'
      assert content_type
      method = 'POST'
    else:
      content_type = None
      method = 'GET'

    if do_deflate and content_type:
      post_data = zlib.compress(post_data, 3)
      content_encoding_header = 'Content-Encoding: deflate\r\n'
    else:
      content_encoding_header = ''

    if content_type:
      content_header = 'Content-Type: %s\r\nContent-Length: %s\r\n' % (
          content_type, len(post_data))
    else:
      content_header = ''

    test_user_header = 'X-Test-User: test@unittest.org\r\n'

    req = '%s %s HTTP/1.0\r\n%s%s%s\r\n%s' % (
        method, suburl, content_header, test_user_header,
        content_encoding_header, post_data or '')
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    lines = []
    rest = None
    f = None
    try:
      try:
        sock.connect(self.address)
        sock.sendall(req)
        f = sock.makefile('r')
        while True:
          line = f.readline()
          assert line, 'EOF in HTTP response header'
          assert line.endswith('\n')
          line = line.strip('\r\n')
          if not line:
            break
          lines.append(line)
        rest = f.read()
      except socket.error, e:
        assert 0, 'socket error to %r: %r' % (self.address, e)
    finally:
      if f:
        f.close()
      sock.close()
    assert lines, 'No first line in HTTP response.'
    line = lines.pop(0)
    match = re.match(r'HTTP/1[.][01] +([2-5][0-9][0-9]) +(\S.*)\Z', line)
    assert match, 'Invalid first line in HTTP response: %r' % line
    response = Response()
    response.status = int(match.group(1))
    response.status_message = match.group(2)
    for line in lines:
      match = re.match(r'([A-Za-z][-A-Za-z0-9]*):[ \t]*(\S.*)\Z', line)
      assert match, 'Invalid HTTP response header: %r' % line
      key = match.group(1).lower()
      value = match.group(2).strip()
      if key == 'content-type':
        match = re.search(r'(?i);\s+charset=', value)
        if match:
          response.content_type = value[:match.start()]
          response.charset = value[match.end():].lower()
        else:
          response.content_type = value
          response.charset = None
      elif key == 'content-length':
        res_content_length = int(value)
      elif key == 'content-encoding':
        if value == 'deflate':
          response.is_compressed = True
        else:
          assert 0, 'Unsupported Content-Encoding value: %r' % value
      elif key == 'x-content-encoding':
        if value == 'deflate':
          response.is_compressed = True
        else:
          assert 0, 'Unexpected X-Content-Encoding value: %r' % value
      elif key == 'transfer-encoding':
        assert 0, 'Unexpected Transfer-Encoding value: %r' % value
      elif key == 'location':
        response.redirect_target = simplify_url(value, self.address)
    if res_content_length is not None:
      assert len(rest) == res_content_length
    assert response.content_type is not None
    if response.is_compressed:
      response.compressed_size = len(rest)
      response.body = zlib.decompress(rest)
    else:
      response.body = rest
    if response.content_type.lower().endswith('/json'):
      try:
        # This creates a JSON with `unicode' objects (not `str' objects), also
        # in dict keys.
        json_unicode = json.loads(
            response.body, encoding=(response.charset or 'ascii'))
      except json.JSONDecodeError, e:
        json_unicode = e
      response.json = recursive_unicode_to_utf8(json_unicode)
    else:
      response.json = NoJson
    return response

client = None


class SmsTest(unittest.TestCase):
  #@Override unittest.TestCase
  def setUp(self):
    assert client
    self.client = client

  def delete_all(self, do_clear_cache=False):
    response = self.client.call(
        '/deleteall' + '?clearcache=1' * bool(do_clear_cache))
    self.assertEqual(302, response.status, (response.status, response.body))
    self.assertEqual('/list', response.redirect_target)
    self.assertEqual(None, self.dump())  # No messages right after delete.

  def dump(self, do_simplify=False):
    """Returns the list of messages in the webapp for the test user."""
    response = self.client.call('/sync?dump=1')
    self.assertEqual(200, response.status)
    self.assertNotEqual(NoJson, response.json)  # Got JSON.
    if response.json is not None:
      for message in response.json:
        assert_full_message(message)
      if do_simplify:
        return map(simplify_message, response.json)
    return response.json

  def sync(self, messages):
    assert isinstance(messages, list)
    req = {'phone': 'test', 'messages': messages}
    response = self.client.call('/sync', post_json=req)
    self.assertEqual(200, response.status, (response.status, response.body))
    self.assertNotEqual(NoJson, response.json)
    self.assertEqual(dict, type(response.json))
    self.assertTrue('smessages' in response.json)
    self.assertEqual(list, type(response.json['smessages']))
    return response.json['smessages']

  def sync_error(self, messages, expected_status, expected_body):
    if isinstance(messages, Response):
      response = messages
    else:
      req = {'phone': 'test', 'messages': messages}
      response = self.client.call('/sync', post_json=req)
    self.assertEqual(expected_status, response.status,
                     (expected_status, response.status, response.body))
    if expected_body is not None:
      self.assertEqual(expected_body, response.body,
                       (expected_body, response.status, response.body))
    if response.status == 200:
      self.assertNotEqual(NoJson, response.json)
    else:
      self.assertEqual(NoJson, response.json)
    return response.json

  def testMainPage(self):
    response = self.client.call('/')
    self.assertEqual(200, response.status)
    self.assertEqual('text/html', response.content_type)
    self.assertEqual('utf-8', response.charset)
    self.assertFalse('HTTP/' in response.body)
    self.assertTrue('>Login<' in response.body)  # Not logged in by default.
    self.assertFalse('>Logout<' in response.body)
    self.assertTrue('="text/html; charset=utf-8"' in response.body)
    self.assertTrue('<p>SyncStar lets you ' in response.body)
    self.assertEqual(NoJson, response.json)

  def testMissingPage(self):
    response = self.client.call('/a/missing/page')
    self.assertEqual(404, response.status)

  def testListPage(self):
    response = self.client.call('/list')
    self.assertEqual(403, response.status)
    self.assertTrue('You have to be logged in to list messages' in
                    response.body)
    response = self.client.call('/list?autologin=1')
    self.assertEqual(302, response.status)
    self.assertTrue((response.redirect_target or '').startswith('/_ah/login?'), response)
    # TODO: Write more tests.

  def testSyncGet(self):
    response = self.client.call('/sync')
    self.assertEqual(200, response.status)
    self.assertEqual('text/html', response.content_type)
    self.assertEqual('utf-8', response.charset)
    self.assertTrue('use your Android device to POST' in response.body)

  def testSyncSyntaxError(self):
    # No need to call self.delete_all(), because in this test we just send
    # bad requests, which just fail without data.
    self.sync_error(self.client.call('/sync', post_data=' \r\t'),
                    400, 'Received empty post data. '
                    'Please send request in JSON format.\n')
    self.sync_error(self.client.call('/sync', post_data='Hello'),
                    400, 'Expected messages in JSON format.\n'
                    'No JSON object could be decoded: '
                    'line 1 column 0 (char 0)\n')
    self.sync_error(self.client.call('/sync', post_data='42'),
                    400, 'Expected dict.\n')
    self.sync_error(self.client.call('/sync', post_data='{}'),
                    400, 'Expected string in phone.\n')
    self.sync_error(self.client.call('/sync', post_data='{}[]true'),
                    400, 'Expected messages in JSON format.\nExtra data: '
                    'line 1 column 2 - line 1 column 8 (char 2 - 8)\n')
    self.sync_error(self.client.call('/sync', post_data='{"answer":42}'),
                    400, 'Expected string in phone.\n')
    self.sync_error(self.client.call('/sync', post_data='{"phone":42}'),
                    400, 'Expected string in phone.\n')
    self.sync_error(self.client.call('/sync', post_data='{"phone":""}'),
                    400, 'Expected non-empty string in phone.\n')
    self.sync_error(self.client.call('/sync', post_data='{"phone":"x"}'),
                    400, 'Expected list in messages.\n')
    self.sync_error(42, 400, 'Expected list in messages.\n')
    self.sync_error({}, 400, 'Expected list in messages.\n')
    self.sync_error([[]], 400, 'Expected list of dict in messages.\n')
    message = {}
    self.sync_error([message], 400, 'Missing address.\n')
    message['address'] = 42
    self.sync_error([message], 400, 'Expected string in address.\n')
    message['address'] = ''
    self.sync_error([message], 400, 'Missing body.\n')
    message['body'] = 42
    self.sync_error([message], 400, 'Expected string in body.\n')
    message['body'] = ''
    self.sync_error([message], 400, 'Missing date.\n')
    message['date'] = '42z'
    self.sync_error([message], 400, 'Expected integer in date.\n')
    message['date'] = '0'
    self.sync_error([message], 400, 'Missing type.\n')
    message['type'] = '-987654'
    self.sync_error([message], 400, 'Expected small integer in type.\n')
    message['type'] = '5'
    self.sync_error([message], 200, None)
    message['deleted'] = '1'
    self.sync_error([message], 400, 'Expected boolean in deleted.\n')
    message['deleted'] = False
    message['status'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in status.\n')
    message['status'] = 42
    self.sync_error([message], 200, None)
    message['error_code'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in error_code.\n')
    message['error_code'] = 42
    self.sync_error([message], 200, None)
    message['locked'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in locked.\n')
    message['locked'] = 42
    self.sync_error([message], 200, None)
    message['read'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in read.\n')
    message['read'] = 42
    self.sync_error([message], 200, None)
    message['seen'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in seen.\n')
    message['seen'] = 42
    self.sync_error([message], 200, None)
    message['protocol'] = 'bad'
    self.sync_error([message], 400, 'Expected small integer in protocol.\n')
    message['protocol'] = 42
    self.sync_error([message], 200, None)
    message['service_center'] = 42
    self.sync_error([message], 400, 'Expected string in service_center.\n')
    message['service_center'] = 'good'
    self.sync_error([message], 200, None)
    message['conversation'] = 42
    self.sync_error([message], 400, 'Expected string in conversation.\n')
    message['conversation'] = 'good'
    self.sync_error([message], 200, None)
    message['whatever'] = [{'blah': 42}]
    self.sync_error([message], 200, None)

  def testSyncData(self):
    self.delete_all()
    # No messages posted, no messages downloaded.
    self.assertEqual([], self.sync([]))
    # The database is still empty.
    self.assertEqual(None, self.dump())
    message1 = {'address': 'a', 'body': 'b', 'date': 1, 'type': 1}
    # One message posted, no messages downloaded.
    self.assertEqual([], self.sync([message1]))
    # The database contains one message.
    self.assertEqual([message1], self.dump(do_simplify=True))
    message2 = {'address': '', 'body': '', 'date': 0, 'type': 0}
    dummy_message2 = dict(message2)
    dummy_message2['dummy'] = ['ignored']
    # Second message posted, first message downloaded.
    self.assertEqual([ph(message1)], self.sync([dummy_message2]))
    # The database contains two messages. They are returned in insertion
    # (not lexicographic) order.
    self.assertEqual([message1, {}], self.dump(do_simplify=True))
    message3 = {'address': 'aaa', 'body': 'bbb', 'date': 3, 'type': 3}
    # Third message posted, first two messages downloaded.
    self.assertEqual([ph(message1), ph({})], self.sync([message3]))
    # The database contains three messages.
    self.assertEqual([message1, {}, message3], self.dump(do_simplify=True))
    # Nothing new posted, all three messages downloaded.
    self.assertEqual([ph(message1), ph({}), ph(message3)], self.sync([]))
    # The database still contains three messages.
    self.assertEqual([message1, {}, message3], self.dump(do_simplify=True))
    # The second and the third messages re-updated, the third downloaded.
    self.assertEqual([ph(message1)], self.sync([message2, message3]))
    # The database still contains three messages.
    self.assertEqual([message1, {}, message3], self.dump(do_simplify=True))
    new_message1 = dict(message1)
    new_message1['read'] = 1
    # The change from read=0 to read=1 propagates from the phone the webapp.
    self.assertEqual([], self.sync([new_message1, message2, message3]))
    # The database has message1 with read=1.
    self.assertEqual([new_message1, {}, message3], self.dump(do_simplify=True))
    # The change to read=1 propagates from the webapp to the phone.
    self.assertEqual([ph(new_message1)],
                      self.sync([message1, message2, message3]))
    # The database has message1 with read=1.
    self.assertEqual([new_message1, {}, message3], self.dump(do_simplify=True))
    back_message1 = dict(message1)
    back_message1['read'] = 0
    # The change from read=0 to read=1 propagates from the webapp to the phone.
    self.assertEqual([ph(new_message1)],
                      self.sync([back_message1, message2, message3]))
    # The database has message1 with read=1.
    self.assertEqual([new_message1, {}, message3], self.dump(do_simplify=True))
    new_message3 = dict(message3)
    new_message3['deleted'] = 42
    bool_message3 = dict(message3)
    bool_message3['deleted'] = True
    # The change from read=0 to read=1 propagates from the webapp to the phone,
    # the change to deleted=True propagates from the phone to the webapp.
    self.assertEqual([ph(new_message1)],
                      self.sync([back_message1, message2, new_message3]))
    # The database has message1 with read=1 and message3 with deleted=True
    dump_result = self.dump(do_simplify=True)
    self.assertEqual([new_message1, {}, bool_message3], dump_result)
    self.assertEqual(bool, type(dump_result[2]['deleted']))
    self.assertEqual([ph(new_message1), ph({})], self.sync([]))
    locked_message3 = dict(bool_message3)
    locked_message3['locked'] = 9
    # Change of locked doesn't propagate from the phone to the webapp.
    self.assertEqual([], self.sync([new_message1, message2, locked_message3]))
    # The database hasn't changed, it still doesn't contain locked.
    self.assertEqual([new_message1, {}, bool_message3],
                     self.dump(do_simplify=True))
    seen_message3 = dict(locked_message3)
    seen_message3['seen'] = 3
    # Increase of seen propagates from the phone to the webapp.
    self.assertEqual([], self.sync([new_message1, message2, seen_message3]))
    seen_unlocked_message3 = dict(seen_message3)
    del seen_unlocked_message3['locked']
    # The database has changed, it contains seen=3.
    self.assertEqual([new_message1, {}, seen_unlocked_message3],
                     self.dump(do_simplify=True))
    # Increase of seen propagates from the webapp to the phone.
    self.assertEqual([ph(seen_unlocked_message3)],
                     self.sync([new_message1, message2, locked_message3]))
    # The database hasn't changed, it still contains seen=3.
    self.assertEqual([new_message1, {}, seen_unlocked_message3],
                     self.dump(do_simplify=True))

  def testSyncConversation(self):
    self.delete_all(do_clear_cache=True)
    # No messages posted, no messages downloaded.
    self.assertEqual([], self.sync([]))
    # The database is still empty.
    self.assertEqual(None, self.dump())
    message1 = {'address': 'MyAddr', 'body': 'b', 'date': 1, 'type': 1,
                'conversation': 'MyConv'}
    # One message posted, no messages downloaded.
    self.assertEqual([], self.sync([message1]))
    # The database contains one message.
    self.assertEqual([message1], self.dump(do_simplify=True))
    message2 = {'address': 'MyAddr2', 'body': 'b',
                'date': 1234567890123456789, 'type': 1}
    conv_message2 = dict(message2)
    conv_message2['conversation'] = 'MyAddr2'
    # One message posted, on messagee downloaded.
    self.assertEqual([ph(message1)], self.sync([conv_message2]))
    # The database contains one message, conversation is not # stored.
    self.assertEqual([message1, message2], self.dump(do_simplify=True))
    # Two messages downloaded, second one without the conversation.
    self.assertEqual([ph(message1), ph(message2)], self.sync([]))

  def testSyncDeflate(self):
    # Compressed HTTP POST request body and reply body.
    self.delete_all()
    msg_json = '{"address":"quux%d","body":"b","date":1,"type":1}'
    req_json = '{"phone":"xy","messages":[%s]}' % ','.join(
        msg_json % i for i in xrange(100))
    long_req_size = len(req_json)
    response = self.client.call('/sync', post_data=req_json, do_deflate=True)
    self.assertEqual(200, response.status)
    self.assertEqual('{"smessages":[]}', response.body)
    self.assertFalse(response.is_compressed)  # Response is too small to compress.

    req_json = '{"phone":"xy","messages":[]}'
    response = self.client.call('/sync', post_data=req_json, do_deflate=True)
    self.assertEqual(200, response.status)
    self.assertTrue(response.is_compressed)
    self.assertNotEqual(None, response.json)
    self.assertEqual(100, len(response.json['smessages']))
    self.assertTrue(response.compressed_size < long_req_size / 10)


if __name__ == '__main__':
  client = Client()
  client.check_health()
  unittest.main(argv=[sys.argv[0], '-v'] + sys.argv[1:])
